package com.voice.core.channel.tcp.impl;

import com.voice.core.channel.IOContext;
import com.voice.core.channel.tcp.IOProcessThread;
import com.voice.core.utils.CloseUtil;
import com.voice.core.utils.ThreadFactoryUtil;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.concurrent.*;

public class IOSelectorContext implements IOContext{

    private final ExecutorService inputPool;
    private final ExecutorService outputPool;
    private final IOProcessThread readThread;
    private final IOProcessThread writeThread;

    public IOSelectorContext() throws IOException {
        this(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(), ThreadFactoryUtil.threadFactory("IOSelectorContext inputPool")),
                Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(), ThreadFactoryUtil.threadFactory("IOSelectorContext outPool")));
    }

    public IOSelectorContext(ExecutorService inputPool, ExecutorService outputPool) throws IOException {
        this.inputPool = inputPool;
        this.outputPool = outputPool;
        readThread = new IOProcessThread("IoSelector Read Thread", SelectionKey.OP_READ, (channel, runnable) -> {
            if (!inputPool.isShutdown()) {
                inputPool.execute(() -> {
                    runnable.run();
                    registerInput(channel, runnable);
                });

            }
        });
        writeThread = new IOProcessThread("IoSelector Write Thread", SelectionKey.OP_WRITE, (channel, runnable) -> {
            if (!outputPool.isShutdown()) {
                outputPool.execute(runnable);
            }
        });
        readThread.setPriority(Thread.MAX_PRIORITY);
        readThread.start();
        writeThread.setPriority(Thread.MAX_PRIORITY);
        writeThread.start();
    }

    @Override
    public boolean registerInput(SocketChannel socketChannel, Runnable callback) {
        return readThread.register(socketChannel, callback) != null;
    }

    @Override
    public boolean registerOutput(SocketChannel socketChannel, Runnable callback) {
        return writeThread.register(socketChannel, callback) != null;
    }

    @Override
    public void unRegisterInput(SocketChannel socketChannel) {
        readThread.unRegister(socketChannel);
    }

    @Override
    public void unRegisterOutput(SocketChannel socketChannel) {
        writeThread.unRegister(socketChannel);
    }

    @Override
    public void close() {
        inputPool.shutdown();
        outputPool.shutdown();
        CloseUtil.close(readThread, writeThread);
    }

}
